﻿using System.Collections.Generic;
using System.Diagnostics.Contracts;
using System.Reactive.Linq;

namespace System.Net.Sockets
{
	public static partial class SocketExtensions
	{
#if !SILVERLIGHT
		/// <summary>
		/// Receives data from a bound <see cref="Socket"/> into the specified <paramref name="buffers"/>.
		/// </summary>
		/// <param name="socket">The socket that will receive the data.</param>
		/// <param name="buffers">The storage location for the received data.</param>
		/// <param name="socketFlags">A bitwise combination of the <see cref="SocketFlags"/> values.</param>
		/// <returns>A singleton observable sequence containing the number of bytes that were received.</returns>
		[System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Naming", "CA1726:UsePreferredTerms", MessageId = "Flags",
			Justification = "Name is consistent with the corresponding BCL method.")]
		public static IObservable<int> ReceiveObservable(this Socket socket, IList<ArraySegment<byte>> buffers, SocketFlags socketFlags)
#else
		/// <summary>
		/// Receives data from a bound <see cref="Socket"/> into the specified <paramref name="buffers"/>.
		/// </summary>
		/// <param name="socket">The socket that will receive the data.</param>
		/// <param name="buffers">The storage location for the received data.</param>
		/// <returns>A singleton observable sequence containing the number of bytes that were received.</returns>
		[System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Naming", "CA1726:UsePreferredTerms", MessageId = "Flags",
			Justification = "Name is consistent with the corresponding BCL method.")]
		public static IObservable<int> ReceiveObservable(this Socket socket, IList<ArraySegment<byte>> buffers)
#endif
		{
			Contract.Requires(socket != null);
			Contract.Requires(buffers != null);
			Contract.Ensures(Contract.Result<IObservable<int>>() != null);

#if !SILVERLIGHT
			var invoke = Observable.FromAsyncPattern<IList<ArraySegment<byte>>, SocketFlags, int>(socket.BeginReceive, socket.EndReceive);

			var observable = invoke(buffers, socketFlags);
#else
			var invoke = Observable.FromAsyncPattern<IList<ArraySegment<byte>>, int>(socket.BeginReceive, socket.EndReceive);

			var observable = invoke(buffers);
#endif

			Contract.Assume(observable != null);

			return observable;
		}

#if !SILVERLIGHT
		/// <summary>
		/// Receives data from a bound <see cref="Socket"/> into the specified <paramref name="buffer"/>.
		/// </summary>
		/// <param name="socket">The socket that will receive the data.</param>
		/// <param name="buffer">The storage location for the received data.</param>
		/// <param name="offset">The zero-based position in the <paramref name="buffer"/> at which to store the received data.</param>
		/// <param name="size">The number of bytes to receive.</param>
		/// <param name="socketFlags">A bitwise combination of the <see cref="SocketFlags"/> values.</param>
		/// <returns>A singleton observable sequence containing the number of bytes that were received.</returns>
		[System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Naming", "CA1726:UsePreferredTerms", MessageId = "Flags",
			Justification = "Name is consistent with the corresponding BCL method.")]
		public static IObservable<int> ReceiveObservable(this Socket socket, byte[] buffer, int offset, int size, SocketFlags socketFlags)
#else
		/// <summary>
		/// Receives data from a bound <see cref="Socket"/> into the specified <paramref name="buffer"/>.
		/// </summary>
		/// <param name="socket">The socket that will receive the data.</param>
		/// <param name="buffer">The storage location for the received data.</param>
		/// <param name="offset">The zero-based position in the <paramref name="buffer"/> at which to store the received data.</param>
		/// <param name="size">The number of bytes to receive.</param>
		/// <returns>A singleton observable sequence containing the number of bytes that were received.</returns>
		[System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Naming", "CA1726:UsePreferredTerms", MessageId = "Flags",
			Justification = "Name is consistent with the corresponding BCL method.")]
		public static IObservable<int> ReceiveObservable(this Socket socket, byte[] buffer, int offset, int size)
#endif
		{
			Contract.Requires(socket != null);
			Contract.Requires(buffer != null);
			Contract.Requires(offset >= 0);
			Contract.Requires(offset <= buffer.Length);
			Contract.Requires(size >= 0);
			Contract.Requires(size <= buffer.Length - offset);
			Contract.Ensures(Contract.Result<IObservable<int>>() != null);

#if !SILVERLIGHT
			var invoke = Observable.FromAsyncPattern<byte[], int, int, SocketFlags, int>(socket.BeginReceive, socket.EndReceive);

			var observable = invoke(buffer, offset, size, socketFlags);
#elif WINDOWS_PHONE
			var invoke = Observable.FromAsyncPattern<Tuple<byte[], int>, int, int>(
				(a1Anda2, a3, callback, state) => socket.BeginReceive(a1Anda2.Item1, a1Anda2.Item2, a3, callback, state),
				asyncResult => socket.EndReceive(asyncResult));

			var observable = invoke(Tuple.Create(buffer, offset), size);
#else
			var invoke = Observable.FromAsyncPattern<byte[], int, int, int>(socket.BeginReceive, socket.EndReceive);

			var observable = invoke(buffer, offset, size);
#endif

			Contract.Assume(observable != null);

			return observable;
		}

#if !SILVERLIGHT
		/// <summary>
		/// Receives data from a bound <see cref="Socket"/> into a sequence of <see cref="byte"/> arrays until the host shuts down 
		/// the connection.
		/// </summary>
		/// <remarks>
		/// <alert type="warning">
		/// Unlike many of the other <see cref="ObservableSocket"/> methods, 	<see cref="ReceiveUntilCompleted(Socket,SocketFlags)"/>
		/// returns a <strong>cold</strong> observable.  This is to ensure that observers do not miss any of the byte arrays in the sequence; otherwise, 
		/// buffering would be required for the entire sequence, similar to how <see cref="Observable.FromAsyncPattern(Func{AsyncCallback,object,IAsyncResult},Action{IAsyncResult})"/> 
		/// buffers the first result and then replays it to all subscribers.  Another consequence of the <strong>cold</strong> behavior is that 
		/// subscriptions are not shared, thus every new subscription will perform an independent receive operation on the <paramref name="socket"/>.
		/// </alert>
		/// </remarks>
		/// <param name="socket">The socket that will receive the data.</param>
		/// <param name="socketFlags">A bitwise combination of the <see cref="SocketFlags"/> values.</param>
		/// <returns>An observable sequence of <see cref="byte"/> arrays containing the data that was received.</returns>
		[System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Naming", "CA1726:UsePreferredTerms", MessageId = "Flags",
			Justification = "Name is consistent with the other methods.")]
		public static IObservable<byte[]> ReceiveUntilCompleted(this Socket socket, SocketFlags socketFlags)
#else
		/// <summary>
		/// Receives data from a bound <see cref="Socket"/> into a sequence of <see cref="byte"/> arrays until the host shuts down 
		/// the connection.
		/// </summary>
		/// <remarks>
		/// <alert type="warning">
		/// Unlike many of the other <see cref="ObservableSocket"/> methods, 	<see cref="ReceiveUntilCompleted(Socket)"/>
		/// returns a <strong>cold</strong> observable.  This is to ensure that observers do not miss any of the byte arrays in the sequence; otherwise, 
		/// buffering would be required for the entire sequence, similar to how <see cref="Observable.FromAsyncPattern(Func{AsyncCallback,object,IAsyncResult},Action{IAsyncResult})"/> 
		/// buffers the first result and then replays it to all subscribers.  Another consequence of the <strong>cold</strong> behavior is that 
		/// subscriptions are not shared, thus every new subscription will perform an independent receive operation on the <paramref name="socket"/>.
		/// </alert>
		/// </remarks>
		/// <param name="socket">The socket that will receive the data.</param>
		/// <returns>An observable sequence of <see cref="byte"/> arrays containing the data that was received.</returns>
		[System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Naming", "CA1726:UsePreferredTerms", MessageId = "Flags",
			Justification = "Name is consistent with the other methods.")]
		public static IObservable<byte[]> ReceiveUntilCompleted(this Socket socket)
#endif
		{
			Contract.Requires(socket != null);
			Contract.Ensures(Contract.Result<IObservable<byte[]>>() != null);

			return Observable.Using(
				() => new SocketAsyncEventArgs()
					{
#if !SILVERLIGHT
						SocketFlags = socketFlags
#endif
					},
				eventArgs =>
				{
					Contract.Assume(socket.ReceiveBufferSize >= 0);

					var buffer = new byte[socket.ReceiveBufferSize];

					eventArgs.SetBuffer(buffer, 0, buffer.Length);

					int received = -1;

					var deferredInvoke = Observable.Defer(() => socket.ReceiveObservable(eventArgs));

#if EXPERIMENTAL
					var loop = Observable.While(() => received != 0, deferredInvoke);
#else
					var loop = Observable2.While(() => received != 0, deferredInvoke);
#endif

					return loop
						.Do(e => received = e.BytesTransferred)
						.Where(e => e.BytesTransferred > 0)
						.Select(_ =>
							{
								if (received < buffer.Length)
								{
									var copy = new byte[received];

									Array.Copy(buffer, copy, received);

									return copy;
								}
								else
								{
									return buffer;
								}
							});
				});
		}

#if !SILVERLIGHT
		/// <summary>
		/// Receives data from a bound <see cref="Socket"/> into the specified <paramref name="buffer"/> and stores the 
		/// actual <see cref="EndPoint"/>.
		/// </summary>
		/// <param name="socket">The socket that will receive the data.</param>
		/// <param name="buffer">The storage location for the received data.</param>
		/// <param name="offset">The zero-based position in the <paramref name="buffer"/> at which to store the received data.</param>
		/// <param name="size">The number of bytes to receive.</param>
		/// <param name="socketFlags">A bitwise combination of the <see cref="SocketFlags"/> values.</param>
		/// <param name="remoteEndPoint">An <see cref="EndPoint"/> that represents the source of the data.</param>
		/// <returns>A singleton observable sequence containing a tuple that indicates the number of bytes that were received
		/// and the actual <see cref="EndPoint"/> from which the data was received.</returns>
		[System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Naming", "CA1726:UsePreferredTerms", MessageId = "Flags",
			Justification = "Name is consistent with the corresponding BCL method.")]
		public static IObservable<Tuple<int, EndPoint>> ReceiveFromObservable(this Socket socket, byte[] buffer, int offset, int size, SocketFlags socketFlags, EndPoint remoteEndPoint)
#elif WINDOWS_PHONE
		/// <summary>
		/// Receives data from a bound <see cref="Socket"/> into the specified <paramref name="buffer"/> and stores the 
		/// actual <see cref="EndPoint"/>.
		/// </summary>
		/// <param name="socket">The socket that will receive the data.</param>
		/// <param name="buffer">The storage location for the received data.</param>
		/// <param name="offset">The zero-based position in the <paramref name="buffer"/> at which to store the received data.</param>
		/// <param name="size">The number of bytes to receive.</param>
		/// <param name="remoteEndPoint">An <see cref="EndPoint"/> that represents the source of the data.</param>
		/// <returns>A singleton observable sequence containing a <see cref="SocketResult"/> that indicates the number of bytes that were received
		/// and the actual <see cref="EndPoint"/> from which the data was received.</returns>
		[System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Naming", "CA1726:UsePreferredTerms", MessageId = "Flags",
			Justification = "Name is consistent with the corresponding BCL method.")]
		public static IObservable<SocketResult> ReceiveFromObservable(this Socket socket, byte[] buffer, int offset, int size, EndPoint remoteEndPoint)
#else
		/// <summary>
		/// Receives data from a bound <see cref="Socket"/> into the specified <paramref name="buffer"/> and stores the 
		/// actual <see cref="EndPoint"/>.
		/// </summary>
		/// <param name="socket">The socket that will receive the data.</param>
		/// <param name="buffer">The storage location for the received data.</param>
		/// <param name="offset">The zero-based position in the <paramref name="buffer"/> at which to store the received data.</param>
		/// <param name="size">The number of bytes to receive.</param>
		/// <param name="remoteEndPoint">An <see cref="EndPoint"/> that represents the source of the data.</param>
		/// <returns>A singleton observable sequence containing a tuple that indicates the number of bytes that were received
		/// and the actual <see cref="EndPoint"/> from which the data was received.</returns>
		[System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Naming", "CA1726:UsePreferredTerms", MessageId = "Flags",
			Justification = "Name is consistent with the corresponding BCL method.")]
		public static IObservable<Tuple<int, EndPoint>> ReceiveFromObservable(this Socket socket, byte[] buffer, int offset, int size, EndPoint remoteEndPoint)
#endif
		{
			Contract.Requires(socket != null);
			Contract.Requires(buffer != null);
			Contract.Requires(offset >= 0);
			Contract.Requires(offset <= buffer.Length);
			Contract.Requires(size >= 0);
			Contract.Requires(size <= buffer.Length - offset);
			Contract.Requires(remoteEndPoint != null);
			Contract.Ensures(Contract.Result<IObservable<Tuple<int, EndPoint>>>() != null);

			EndPoint actualRemoteEndPoint = remoteEndPoint;

#if !SILVERLIGHT
			var invoke = Observable.FromAsyncPattern<byte[], int, int, SocketFlags, int>(
				(a1, a2, a3, a4, callback, state) => socket.BeginReceiveFrom(a1, a2, a3, a4, ref actualRemoteEndPoint, callback, state),
				asyncResult => socket.EndReceiveFrom(asyncResult, ref actualRemoteEndPoint));

			var observable = invoke(buffer, offset, size, socketFlags);

			Contract.Assume(observable != null);

			return observable.Select(count => Tuple.Create(count, actualRemoteEndPoint));
#elif WINDOWS_PHONE
			var invoke = Observable.FromAsyncPattern<Tuple<byte[], int>, int, int>(
				(a1Anda2, a3, callback, state) => socket.BeginReceiveFrom(a1Anda2.Item1, a1Anda2.Item2, a3, ref actualRemoteEndPoint, callback, state),
				asyncResult => socket.EndReceiveFrom(asyncResult, ref actualRemoteEndPoint));

			var observable = invoke(Tuple.Create(buffer, offset), size);

			Contract.Assume(observable != null);

			return observable.Select(count => new SocketResult(count, actualRemoteEndPoint));
#else
			var invoke = Observable.FromAsyncPattern<byte[], int, int, int>(
				(a1, a2, a3, callback, state) => socket.BeginReceiveFrom(a1, a2, a3, ref actualRemoteEndPoint, callback, state),
				asyncResult => socket.EndReceiveFrom(asyncResult, ref actualRemoteEndPoint));

			var observable = invoke(buffer, offset, size);

			Contract.Assume(observable != null);

			return observable.Select(count => Tuple.Create(count, actualRemoteEndPoint));
#endif
		}

#if !SILVERLIGHT
		/// <summary>
		/// Receives data from a bound <see cref="Socket"/> into the specified <paramref name="buffer"/> and stores the 
		/// actual <see cref="SocketFlags"/>, <see cref="EndPoint"/> and packet information.
		/// </summary>
		/// <param name="socket">The socket that will receive the data.</param>
		/// <param name="buffer">The storage location for the received data.</param>
		/// <param name="offset">The zero-based position in the <paramref name="buffer"/> at which to store the received data.</param>
		/// <param name="size">The number of bytes to receive.</param>
		/// <param name="socketFlags">A bitwise combination of the <see cref="SocketFlags"/> values.</param>
		/// <param name="remoteEndPoint">An <see cref="EndPoint"/> that represents the source of the data.</param>
		/// <returns>A singleton observable sequence containing a tuple that indicates the number of bytes that were received
		/// and the actual <see cref="EndPoint"/> from which the data was received.</returns>
		[System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Naming", "CA1726:UsePreferredTerms", MessageId = "Flags",
			Justification = "Name is consistent with the corresponding BCL method.")]
		public static IObservable<Tuple<int, SocketFlags, EndPoint, IPPacketInformation>> ReceiveMessageFromObservable(this Socket socket, byte[] buffer, int offset, int size, SocketFlags socketFlags, EndPoint remoteEndPoint)
		{
			Contract.Requires(socket != null);
			Contract.Requires(buffer != null);
			Contract.Requires(offset >= 0);
			Contract.Requires(offset <= buffer.Length);
			Contract.Requires(size >= 0);
			Contract.Requires(size <= buffer.Length - offset);
			Contract.Requires(remoteEndPoint != null);
			Contract.Ensures(Contract.Result<IObservable<Tuple<int, SocketFlags, EndPoint, IPPacketInformation>>>() != null);

			EndPoint actualRemoteEndPoint = remoteEndPoint;
			SocketFlags actualSocketFlags = socketFlags;

			var packet = default(IPPacketInformation);

			var invoke = Observable.FromAsyncPattern<byte[], int, int, SocketFlags, int>(
				(a1, a2, a3, a4, callback, state) => socket.BeginReceiveMessageFrom(a1, a2, a3, a4, ref actualRemoteEndPoint, callback, state),
				asyncResult => socket.EndReceiveMessageFrom(asyncResult, ref actualSocketFlags, ref actualRemoteEndPoint, out packet));

			var observable = invoke(buffer, offset, size, socketFlags);

			Contract.Assume(observable != null);

			return observable.Select(count => Tuple.Create(count, actualSocketFlags, actualRemoteEndPoint, packet));
		}
#endif

		/// <summary>
		/// Receives data from a bound <see cref="Socket"/>.
		/// </summary>
		/// <param name="socket">The socket that will receive the data.</param>
		/// <param name="eventArgs">The <see cref="SocketAsyncEventArgs"/> object to use for this asynchronous socket operation.</param>
		/// <returns>A singleton observable sequence containing the result of the operation.</returns>
		/// <seealso href="http://msdn.microsoft.com/en-us/library/system.net.sockets.socket.receiveasync.aspx">Socket.ReceiveAsync</seealso>
		public static IObservable<SocketAsyncEventArgs> ReceiveObservable(this Socket socket, SocketAsyncEventArgs eventArgs)
		{
			Contract.Requires(socket != null);
			Contract.Requires(eventArgs != null);
			Contract.Ensures(Contract.Result<IObservable<SocketAsyncEventArgs>>() != null);

			return eventArgs.InvokeAsync(socket.ReceiveAsync);
		}

		/// <summary>
		/// Receives data from a bound <see cref="Socket"/>.
		/// </summary>
		/// <param name="socket">The socket that will receive the data.</param>
		/// <param name="eventArgs">The <see cref="SocketAsyncEventArgs"/> object to use for this asynchronous socket operation.</param>
		/// <returns>A singleton observable sequence containing the result of the operation.</returns>
		/// <seealso href="http://msdn.microsoft.com/en-us/library/system.net.sockets.socket.receivefromasync.aspx">Socket.ReceiveFromAsync</seealso>
		public static IObservable<SocketAsyncEventArgs> ReceiveFromObservable(this Socket socket, SocketAsyncEventArgs eventArgs)
		{
			Contract.Requires(socket != null);
			Contract.Requires(eventArgs != null);
			Contract.Ensures(Contract.Result<IObservable<SocketAsyncEventArgs>>() != null);

			return eventArgs.InvokeAsync(socket.ReceiveFromAsync);
		}

#if !SILVERLIGHT
		/// <summary>
		/// Receives data from a bound <see cref="Socket"/>.
		/// </summary>
		/// <param name="socket">The socket that will receive the data.</param>
		/// <param name="eventArgs">The <see cref="SocketAsyncEventArgs"/> object to use for this asynchronous socket operation.</param>
		/// <returns>A singleton observable sequence containing the result of the operation.</returns>
		/// <seealso href="http://msdn.microsoft.com/en-us/library/system.net.sockets.socket.receivemessagefromasync.aspx">Socket.ReceiveMessageFromAsync</seealso>
		public static IObservable<SocketAsyncEventArgs> ReceiveMessageFromObservable(this Socket socket, SocketAsyncEventArgs eventArgs)
		{
			Contract.Requires(socket != null);
			Contract.Requires(eventArgs != null);
			Contract.Ensures(Contract.Result<IObservable<SocketAsyncEventArgs>>() != null);

			return eventArgs.InvokeAsync(socket.ReceiveMessageFromAsync);
		}
#endif
	}
}